package za.org.jcicada.katcp;

/**
 * Client.java
 *
 * @author Simon Cross
 *
 * Created on October 13, 2008, 13:45
 */

import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.nio.charset.CharsetEncoder;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * Represents a connection to a KATCP device.
 *
 * Subclasses should implement .reply() and .inform() to
 * take actions when reply and inform message arrive.
 *
 * Request messages can be sent by calling .request().
 */
public abstract class Client implements Runnable {

    private static Log logger = LogFactory.getLog(Client.class);

    /**
     * Decoder for decoding incoming message strings.
     */
    protected static CharsetDecoder wireDecoder = Charset.forName("ascii").newDecoder();

    /**
     * Encoder for encoding outgoing message strings.
     */
    protected static CharsetEncoder wireEncoder = Charset.forName("ascii").newEncoder();

    /**
     * Whether the client is ready to run (i.e. not in the process of stopping).
     */
    // canRun is kept true except while the client is stopping
    protected boolean canRun = true;

    /**
     * Flag that determines whether the client is currently running.
     */
    protected boolean running = false;

    /**
     * Parser instance for parsing strings into Message objects.
     */
    protected Parser parser = new Parser();

    /**
     * Channel to server.
     */
    protected SocketChannel sc = null;

    /**
     * Host to connect to.
     */
    protected String host;

    /**
     * Port to connect to.
     */
    protected int port;

    /**
     * MessageSendException (so we can distinguish our IOErrors
     * from the errors of others).
     */
    public static class MessageSendException extends IOException {
        private static final long serialVersionUID = 1L;

        /**
         * Constructs a MessageSendException object.
         *
         * @param message  The reason for the exception.
         * @param cause  The exception that caused the send to fail.
         */
        public MessageSendException (String message, Throwable cause) {
            super(message);
            initCause(cause);
        }
    }

    /**
     * Create a KATCP client.
     *
     * @param host  The host to connect to.
     * @param port  The port to connect to.
     */
    public Client(String host, int port) {
        this.host = host;
        this.port = port;
    }

    /**
     * Called when a reply message arrives.
     *
     * @param msg  The reply message received.
     */
    public abstract void reply(Message msg);

    /**
     * Called when an inform message arrives.
     *
     * @param msg  The inform message received.
     */
    public abstract void inform(Message msg);

    /**
     * Send a message to the server.
     *
     * @param msg  The Message to send to the server.
     * @throws MessageSendException  If the send fails.
     */
    public void sendMessage(Message msg)
    throws MessageSendException {
        try {
            ByteBuffer buf = wireEncoder.encode(CharBuffer.wrap(msg.toString() + "\n"));
            sc.write(buf);
        }
        catch (IOException e) {
            throw new MessageSendException("Failed to send message.", e);
        }
    }

    /**
     * Send a request message.
     *
     * @param msg  The request Message to send to the server.
     * @throws MessageSendException  If the send fails.
     */
    public void request(Message msg)
    throws MessageSendException {
        if (msg.getType() != Message.Type.REQUEST)
            throw new RuntimeException("Client.request() should only be used"
                                     + " to send request messages.");
        sendMessage(msg);
    }

    /**
     * Connect or reconnect to the server.
     *
     * @throws IOException if the connect fails.
     */
    public void connect()
    throws java.io.IOException {
        if (sc != null) {
            sc.close();
        }
        sc = SocketChannel.open(
            new InetSocketAddress(host, port)
        );
        sc.configureBlocking(false);
    }

    /**
     * Handle a chunk of data from the server.
     *
     * @param chunk  The data from the server to process.
     * @return Any unprocessed data that remains.
     */
    public String handleChunk(String chunk) {
        chunk = chunk.replace('\r', '\n');
        String [] lines = chunk.split("\\n", -1);
        String fullLine;
        Message msg;

        for (int i=0; i < lines.length-1; i++) {
            fullLine = lines[i];
            try {
                msg = parser.parse(fullLine);
                handleMessage(msg);
            }
            catch (Exception e) {
                // usually the parser will throw Parser.KatcpSyntaxExceptions
                // but we really do want to catch other errors too.
                StringWriter sw = new StringWriter();
                e.printStackTrace(new PrintWriter(sw));
                logger.error(String.format("Error handling message from server ['%s']", sw));
            }
        }

        return lines[lines.length-1];        
    }

    /**
     * Handle a message from the server.
     *
     * @param msg  The Message to process.
     */
    public void handleMessage(Message msg) {
        if (msg.getType() == Message.Type.INFORM) {
            inform(msg);
        }
        else if (msg.getType() == Message.Type.REPLY) {
            reply(msg);
        }
        else {
            logger.error(String.format("Unexpected message type from server ['%s']", msg));
        }
    }

    /**
     * Process reply and inform messages from the server.
     */
    public void run() {
        Selector selector;
        SocketChannel tmpChannel;
        ByteBuffer buffer = ByteBuffer.allocate(4096);
        CharBuffer cbuffer = CharBuffer.allocate(4096);
        String chunk = "";
        long timeout = 50; // select timeout in ms

        try {
            connect();
            selector = Selector.open();
            sc.register(selector, SelectionKey.OP_READ);
        }
        catch (IOException e) {
            throw new RuntimeException("Failed to start device client", e);
        }

        synchronized (this) {
            running = canRun;
        }

        while (running) {
            try {
                selector.select(timeout);
            }
            catch (IOException e) {
                throw new RuntimeException("Client selector threw an IOException", e);
            }

            for (SelectionKey key: selector.selectedKeys()) {
                if ((key.readyOps() & SelectionKey.OP_READ) != 0) {
                    tmpChannel = (SocketChannel) key.channel();

                    buffer.clear();
                    try {
                        tmpChannel.read(buffer);
                    }
                    catch (IOException e) {
                        try {
                            connect();
                            sc.register(selector, SelectionKey.OP_READ);
                        }
                        catch (IOException closeExp) {
                            StringWriter sw = new StringWriter();
                            closeExp.printStackTrace(new PrintWriter(sw));
                            logger.error(String.format("Failed to reconnect to server ['%s']", sw));
                            throw new RuntimeException("Failed to reconnect to server", closeExp);
                        }
                        continue;
                    }

                    buffer.flip();
                    if (buffer.limit() == 0) {
                        // read flag with no data == server closed
                        try {
                            connect();
                            sc.register(selector, SelectionKey.OP_READ);
                        }
                        catch (IOException closeExp) {
                            StringWriter sw = new StringWriter();
                            closeExp.printStackTrace(new PrintWriter(sw));
                            logger.error(String.format("Failed to reconnect to server ['%s']", sw));
                            throw new RuntimeException("Failed to reconnect to server", closeExp);
                        }
                        continue;
                    }

                    cbuffer.clear();
                    wireDecoder.decode(buffer, cbuffer, true);
                    cbuffer.flip();
                    chunk = handleChunk(chunk + cbuffer.toString());
                }
            }

            selector.selectedKeys().clear();
        }

        if (sc != null) { 
            try { sc.close(); }
            catch (IOException e) {}
            sc = null;
        }

        // prepare for next call to run()
        synchronized (this) {
            canRun = true;
        }
    }

    /**
     * Check whether a client is running.
     *
     * @return True if the client is running, false otherwise.
     */
    public synchronized boolean isRunning() {
        return running;
    }

    /**
     * Stop a running client (from another thread).
     */
    public synchronized void stop() {
        running = false;
        canRun = false;
    }
}
